/**
 * Copyright (C) 2008 by The Regents of the University of California
 * Redistribution of this file is permitted under the terms of the GNU
 * Public License (GPL).
 *
 * @author Junghoo "John" Cho <cho AT cs.ucla.edu>
 * @date 3/24/2008
 */

#include <cstdio>
#include <iostream>
#include <fstream>
#include "Bruinbase.h"
#include "SqlEngine.h"
#include "BTreeIndex.h"

using namespace std;

#define CS143_TRY(COMMAND)              \
do {                                    \
    RC retVal = COMMAND;                \
  if (retVal != 0) { return retVal; } \
} while (0)

// external functions and variables for load file and sql command parsing
extern FILE* sqlin;
int sqlparse(void);


RC SqlEngine::run(FILE* commandline)
{
  fprintf(stdout, "Bruinbase> ");

  // set the command line input and start parsing user input
  sqlin = commandline;
  sqlparse();  // sqlparse() is defined in SqlParser.tab.c generated from
               // SqlParser.y by bison (bison is GNU equivalent of yacc)

  return 0;
}

RC SqlEngine::select(int attr, const string& table, const vector<SelCond>& cond)
{
  RecordFile  rf;   // RecordFile containing the table
  RecordId    rid;  // record cursor for table scanning
  IndexCursor cur;  // index cursor
  BTreeIndex  idx;  // index file

  RC     rc;
  int    key;
  string value;
  int    count;
  int    diff;
  bool   hasIndex = false;
  RangePair range;
  vector<int> holes;
  int scanMethod = SCAN_METHOD_NO_INDEX; // default to no index
  bool readRecordFile = true; // default to read rf

  // open the table file
  if ((rc = rf.open(table + ".tbl", 'r')) < 0) {
    fprintf(stderr, "Error: table %s does not exist\n", table.c_str());
    return rc;
  }

  // open the index file
  hasIndex = (idx.open(table + ".idx", 'r') == 0);

  scanMethod = chooseScanMethod(cond, range, holes, readRecordFile, attr);

  // use index if valid subranges were found
  if (scanMethod == SCAN_METHOD_USE_INDEX && hasIndex) {
    count = 0;

    // single value
    if (range.low == range.high) {
      // locate and readforward
      CS143_TRY(idx.locate(range.low, cur));

      int retrievedKey;
      RecordId retrievedRid;

      CS143_TRY(idx.readForward(cur, retrievedKey, retrievedRid));
      CS143_TRY(rf.read(retrievedRid, retrievedKey, value));

      if (retrievedKey != range.low) // range.low == range.high
        goto finish_query;

      // check "value" conditions
      for (int i = 0; i < cond.size(); ++i) {
        if (cond[i].attr != 2)
          continue;

        diff = strcmp(value.c_str(), cond[i].value);

        // skip the tuple if any condition is not met
        switch (cond[i].comp) {
          case SelCond::EQ:
            if (diff != 0) goto finish_query;
            break;
          case SelCond::NE:
            if (diff == 0) goto finish_query;
            break;
          case SelCond::GT:
            if (diff <= 0) goto finish_query;
            break;
          case SelCond::LT:
            if (diff >= 0) goto finish_query;
            break;
          case SelCond::GE:
            if (diff < 0) goto finish_query;
            break;
          case SelCond::LE:
            if (diff > 0) goto finish_query;
            break;
        }
      }

      // the condition is met for the tuple.
      // increase matching tuple counter
      count++;

      // print the tuple
      switch (attr) {
        case 1:  // SELECT key
          fprintf(stdout, "%d\n", retrievedKey);
          break;
        case 2:  // SELECT value
          fprintf(stdout, "%s\n", value.c_str());
          break;
        case 3:  // SELECT *
          fprintf(stdout, "%d '%s'\n", retrievedKey, value.c_str());
          break;
      }
    }

    // continuous range. scan [range.low, range.high] and skip "<>"
    else {
      CS143_TRY(idx.locate(range.low, cur));

      int retrievedKey;
      RecordId retrievedRid;

      while (idx.readForward(cur, retrievedKey, retrievedRid) == 0) {
        // only read values when necessary
        if (readRecordFile) {
          CS143_TRY(rf.read(retrievedRid, retrievedKey, value));
        }

        // if we passed the upper bound
        if (retrievedKey > range.high)
          break;

        // ignore "<>"
        for (int i = 0; i < holes.size(); ++i) {
          if (retrievedKey == holes[i])
            goto read_next;
        }

        // check "value" conditions
        if (readRecordFile) {
          for (int i = 0; i < cond.size(); ++i) {
            if (cond[i].attr != 2)
              continue;

            diff = strcmp(value.c_str(), cond[i].value);

            // skip the tuple if any condition is not met
            switch (cond[i].comp) {
              case SelCond::EQ:
                if (diff != 0) goto read_next;
                break;
              case SelCond::NE:
                if (diff == 0) goto read_next;
                break;
              case SelCond::GT:
                if (diff <= 0) goto read_next;
                break;
              case SelCond::LT:
                if (diff >= 0) goto read_next;
                break;
              case SelCond::GE:
                if (diff < 0) goto read_next;
                break;
              case SelCond::LE:
                if (diff > 0) goto read_next;
                break;
            }
          }
        }

        // the condition is met for the tuple.
        // increase matching tuple counter
        count++;

        // print the tuple
        switch (attr) {
          case 1:  // SELECT key
            fprintf(stdout, "%d\n", retrievedKey);
            break;
          case 2:  // SELECT value
            fprintf(stdout, "%s\n", value.c_str());
            break;
          case 3:  // SELECT *
            fprintf(stdout, "%d '%s'\n", retrievedKey, value.c_str());
            break;
        }

        read_next:;
      }
    }

    goto finish_query;
  }

  // scan without index
  else if (scanMethod == SCAN_METHOD_NO_INDEX || !hasIndex) {

    // scan the table file from the beginning
    rid.pid = rid.sid = 0;
    count = 0;

    while (rid < rf.endRid()) {
      // read the tuple
      if ((rc = rf.read(rid, key, value)) < 0) {
        fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
        goto exit_select;
      }

      // check the conditions on the tuple
      for (unsigned i = 0; i < cond.size(); i++) {
        // compute the difference between the tuple value and the condition value
        switch (cond[i].attr) {
          case 1:
            diff = key - atoi(cond[i].value);
            break;
          case 2:
            diff = strcmp(value.c_str(), cond[i].value);
            break;
        }

        // skip the tuple if any condition is not met
        switch (cond[i].comp) {
          case SelCond::EQ:
            if (diff != 0) goto next_tuple;
            break;
          case SelCond::NE:
            if (diff == 0) goto next_tuple;
            break;
          case SelCond::GT:
            if (diff <= 0) goto next_tuple;
            break;
          case SelCond::LT:
            if (diff >= 0) goto next_tuple;
            break;
          case SelCond::GE:
            if (diff < 0) goto next_tuple;
            break;
          case SelCond::LE:
            if (diff > 0) goto next_tuple;
            break;
        }
      }

      // the condition is met for the tuple.
      // increase matching tuple counter
      count++;

      // print the tuple
      switch (attr) {
        case 1:  // SELECT key
          fprintf(stdout, "%d\n", key);
          break;
        case 2:  // SELECT value
          fprintf(stdout, "%s\n", value.c_str());
          break;
        case 3:  // SELECT *
          fprintf(stdout, "%d '%s'\n", key, value.c_str());
          break;
      }

      // move to the next tuple
      next_tuple:
      ++rid;
    }
  }

  // no scanning
  else if (scanMethod == SCAN_METHOD_NO_SCAN) {}

  finish_query:

  // print matching tuple count if "select count(*)"
  if (attr == 4) {
    fprintf(stdout, "%d\n", count);
  }
  rc = 0;

  // close the table file and return
  exit_select:
  rf.close();
  return rc;
}

RC SqlEngine::load(const string& table, const string& loadfile, bool index)
{
  string line;
  int key;
  string value;

  RC         rc;  // return value of RecordFile::open
  RecordFile rf;  // RecordFile containing the table
  RecordId   rid; // record cursor for table scanning
  BTreeIndex idx; // index file

  // create new table file for writing
  if ((rc = rf.open(table + ".tbl", 'w')) < 0) {
    fprintf(stderr, "Error: cannot open table '%s' for writing\n", table.c_str());
    return rc;
  }

  // create index file if index is true
  if (index && (rc = idx.open(table + ".idx", 'w')) < 0) {
    fprintf(stderr, "Error: cannot create index file '%s.idx'\n", table.c_str());
    return rc;
  }

  // file I/O
  ifstream fin(loadfile.c_str());

  if (fin.good()) {
    // initialize cursors
    rid.pid = rid.sid = 0;

    // fetch each line and append the key and value to the RecordFile
    while (getline(fin, line)) {
      parseLoadLine(line, key, value);
      rf.append(key, value, rid);
      if (index) {
        idx.insert(key, rid);
      }
    }
  }
  else {
    fprintf(stderr, "Error: cannot load file '%s'\n", loadfile.c_str());
    return RC_FILE_READ_FAILED;
  }

  return 0;
}

RC SqlEngine::parseLoadLine(const string& line, int& key, string& value)
{
    const char *s;
    char        c;
    string::size_type loc;

    // ignore beginning white spaces
    c = *(s = line.c_str());
    while (c == ' ' || c == '\t') { c = *++s; }

    // get the integer key value
    key = atoi(s);

    // look for comma
    s = strchr(s, ',');
    if (s == NULL) { return RC_INVALID_FILE_FORMAT; }

    // ignore white spaces
    do { c = *++s; } while (c == ' ' || c == '\t');

    // if there is nothing left, set the value to empty string
    if (c == 0) {
        value.erase();
        return 0;
    }

    // is the value field delimited by ' or "?
    if (c == '\'' || c == '"') {
        s++;
    } else {
        c = '\n';
    }

    // get the value string
    value.assign(s);
    loc = value.find(c, 0);
    if (loc != string::npos) { value.erase(loc); }

    return 0;
}

RC SqlEngine::chooseScanMethod(const vector<SelCond>& cond,
                               RangePair& range,
                               vector<int>& holes,
                               bool& readRecordFile,
                               int attr) {

  vector<SelCond> greater,
                  less,
                  equal,
                  notequal;

  int whereKeyCount = 0;
  int whereValCount = 0;

  range.low  = INT_MIN;
  range.high = INT_MAX;


  // populate respective vectors
  for (int i = 0; i < cond.size(); ++i) {

    // check conditions on keys
    if (cond[i].attr == 1) {
      whereKeyCount++;

      SelCond tempCond = cond[i];
      int maxIntCharWidth = 12;

      switch (cond[i].comp) {
        case SelCond::GT:
          snprintf(tempCond.value, maxIntCharWidth, "%d", atoi(tempCond.value) + 1);
        case SelCond::GE:
          greater.push_back(tempCond);
          break;

        case SelCond::LT:
          snprintf(tempCond.value, maxIntCharWidth, "%d", atoi(tempCond.value) - 1);
        case SelCond::LE:
          less.push_back(tempCond);
          break;

        case SelCond::EQ:
          equal.push_back(tempCond);
          break;

        case SelCond::NE:
          notequal.push_back(tempCond);
          break;
      }
    }
    else {
      whereValCount++;
    }
  }

  // logical OR with original value
  // essentially, if: "SELECT *" or "SELECT value" or "SELECT ... WHERE ... value ..."
  readRecordFile = attr == 2 || attr == 3 || (whereValCount > 0);

  // sort them all
  sort(greater.begin(), greater.end());
  sort(less.begin(), less.end());
  sort(equal.begin(), equal.end());
  sort(notequal.begin(), notequal.end());

  // remove duplicates
  greater.erase(unique(greater.begin(), greater.end()), greater.end());
  less.erase(unique(less.begin(), less.end()), less.end());
  equal.erase(unique(equal.begin(), equal.end()), equal.end());
  notequal.erase(unique(notequal.begin(), notequal.end()), notequal.end());

  // update "holes" vector (these are the not-equal values that will be skipped during range scan)
  for (int i = 0; i < notequal.size(); ++i) {
    holes.push_back(atoi(notequal[i].value));
  }

  // update the range boundaries
  if (!greater.empty())
    range.low = atoi(greater.back().value);
  if (!less.empty())
    range.high = atoi(less.front().value);

  // check range validity
  if (!range.isValid())
    return SCAN_METHOD_NO_SCAN;

  // only max one "=" is allowed, and it must be in the range
  if (equal.size() > 1)
    return SCAN_METHOD_NO_SCAN;

  //  "=" must be within the range
  if (equal.size() == 1) {
    int eqKey = atoi(equal[0].value);
    if (eqKey < range.low || eqKey > range.high)
      return SCAN_METHOD_NO_SCAN;
  }

  // update range for "=""
  if (equal.size() == 1) {
    range.low = range.high = atoi(equal[0].value);
  }

  // ignore "<>" that exist outside of the range
  for (int i = 0; i < notequal.size(); ++i) {
    int neqKey = atoi(notequal[i].value);

    // "<>" and "=" on the same key value
    if (equal.size() == 1 && neqKey == atoi(equal[0].value))
      return SCAN_METHOD_NO_SCAN;

    // if range is a single value (not explicitly a "="), and
    // "<>" is the same value
    if (range.low == range.high && range.low == neqKey)
      return SCAN_METHOD_NO_SCAN;
  }

  if (attr == 2 || attr == 3) {
    if (readRecordFile && range.low == INT_MIN && range.high == INT_MAX) {
      return SCAN_METHOD_NO_INDEX;
    }
    else{
      return SCAN_METHOD_USE_INDEX;
    }
  }
  else {
    if (whereValCount > 0 && whereKeyCount == 0) {
      return SCAN_METHOD_NO_INDEX;
    }
    else {
      return SCAN_METHOD_USE_INDEX;
    }
  }
}
